package com.cxdr.awsips.distribute.processor;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.cxdr.awsips.common.logger.LogFactory;
import com.cxdr.awsips.common.thread.BaseProcessor;
import com.cxdr.awsips.distribute.bean.DistributionLog;
import com.cxdr.awsips.distribute.bean.Link_Target;
import com.cxdr.awsips.distribute.bean.ReportFileLog;
import com.cxdr.awsips.distribute.bean.Target;
import com.cxdr.awsips.distribute.config.LinkConfig;
import com.cxdr.awsips.distribute.config.SystemConfig;
import com.cxdr.awsips.distribute.file.FileHelper;
import com.cxdr.awsips.distribute.file.FileHelperFactoy;
import com.cxdr.awsips.distribute.pool.BaseQueue;
import com.cxdr.awsips.distribute.pool.UnCompletePool;
import com.cxdr.awsips.distribute.rule.FileNameRule;
import com.cxdr.awsips.distribute.rule.StationRule;
import com.cxdr.awsips.distribute.rule.TimeRule;

/**
 * 文件分发线程,一个链路对应一个线程池
 * 
 * @author pugang
 * 
 */
public class DistributeProccessor extends BaseProcessor {
	private static Logger logger = LogFactory
			.makeNewLogger(DistributeProccessor.class);
	public static final String THREAD_NAME = "DistributeProccessor";
	public final BaseQueue<File> DistributeFileQueue = new BaseQueue<File>();
	public final String ERROR_RESULT = "-1";
	public final String SUCCESE_RESULT = "1";
	private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM");
	private Link_Target target;
	private LinkConfig link;
	private String sourceID = "1";
	private DistributeThread[] distributeThreads = new DistributeThread[Integer
			.parseInt(SystemConfig.get("num.uploadthread"))];
	private ExecutorService executor = Executors
			.newFixedThreadPool(distributeThreads.length);

	private class DistributeThread extends BaseProcessor {
		private FileHelper targetHelper;
		private static final String THREAD_NAME = "DistributeProccessor-Distributer";
		public DistributeThread() {
			try {
				targetHelper = FileHelperFactoy.getHelper(target);
				targetHelper.open();
				targetHelper.setLocalDir(link.getLocalDir());
				targetHelper.setRemoteDir(target.getFolder());
				setTask_name(DistributeWorker.TASK_NAME);
				setThread_name(THREAD_NAME);
			} catch (IOException e) {
				logger.error("获取远程连接错误 ", e);
				e.printStackTrace();
			}
		}

		@Override
		public void run() {
			String localDirAbsPath = "";
			String sourceAddr = "";
			Target _s = link.getSources().get(0);
			localDirAbsPath= _s.getFolder();
			sourceAddr = _s.getIp();
			while (!isFinish()) {
				File file = DistributeFileQueue.take();
				if (file == null) {
					logger.error("file is null");
					continue;
				}
				logger.info(" 取出文件" + file+"|tid:"+getId());
				DistributionLog log = new DistributionLog(file.getName());
				//设置log
				Date begin_time = new Date();
				String result = "";
				log.setBegin_time(begin_time);
				log.setSource_address(sourceAddr);
				log.setSource_path(localDirAbsPath);
				log.setPartition_month(dateFormat.format(begin_time));
				log.setSource_type(String.valueOf(target.getProtocol().ordinal()));
				log.setTarget_path(target.getFolder());
				log.setTarget_type(String.valueOf(target.getProtocol().ordinal()));
				log.setTartget_address(target.getIp());
				log.setCreate_time(begin_time);
				if (targetHelper.uploadFile(file)) {
					result = SUCCESE_RESULT;
					UnCompletePool.getInstance().add(file.getAbsolutePath(),sourceID, target.getId(),
							SUCCESE_RESULT);
				} else {
					// 如果上传不成功,则写入错误标志
					result = ERROR_RESULT;
					UnCompletePool.getInstance().add(file.getAbsolutePath(),sourceID, target.getId(),
							ERROR_RESULT);
				}
				log.setEnd_time(new Date());
				log.setResult(result);
				StoreDistributeProccessor.DistrbuteFileLogQueue
				.add(log);
				if(DistributeFileQueue.isEmpty()){
					finish();
				}
			}
		}
	}

	public DistributeProccessor(LinkConfig _link,Link_Target _target) {
		link = _link;
		target = _target;
		setTask_name(DistributeWorker.TASK_NAME);
		setThread_name(THREAD_NAME);
		for (int i=0;i<distributeThreads.length;i++) {
			distributeThreads[i] = new DistributeThread();
		}
	}
	
	@Override
	@Deprecated
	public void finish(){
		super.finish();
		
	}
	public void distribute(){
		logger.info("Start DistributeProccessor !");
		for (DistributeThread dt : distributeThreads) {
			executor.submit(dt);
		}
	}
	@Override
	@Deprecated
	public void run() {
		logger.info("Start DistributeProccessor : " + this.getId());
		for (DistributeThread dt : distributeThreads) {
			executor.submit(dt);
		}
//		logger.info("DistributeProccessor :" + this.getId() + " End");
	}
	/**
	 * 按照规则过滤文件,各种文件规则是与的关系，只要有一种规则不满足，那么就筛选出去
	 * @param file
	 * @return
	 */
	public boolean accept(File file){
		for(FileNameRule fileNameRule:target.getFileNameRules()){
			if (fileNameRule != null && !fileNameRule.accept(file.getName())) {
				logger.info("不符合文件名匹配规则："+fileNameRule.desc()+"  ; file name:"+file.getName());
				return false;
			}
		}
		ReportFileLog log = new ReportFileLog();
		log.setFile_name(file.getName(), true);
		for (TimeRule timeRule : target.getTimeRules()) {
			if (timeRule != null
					&& !timeRule.accept(log.getObser_time())) {
				logger.info("不符合时间匹配规则："+timeRule.desc()+"  ; file name:"+file.getName());
				return false;
			}
		}
		for (StationRule staionRule:target.getStationRules()){
			if (staionRule != null
					&& !staionRule.accept(log.getStation_ID())) {
				logger.info("不符合站点匹配规则："+staionRule.desc()+"  ; file name:"+file.getName());
				return false;
			}
		}
		return true;
	}
}
